#include "kvdb/server/kvdbserver.h"

namespace kvdb {

KvdbServer::KvdbServer(io_service_ptr io_service)
  :io_service_(io_service),
   kvdb_sqlite_(new KvdbSqlite()) {
}

KvdbServer::~KvdbServer() {
}

bool KvdbServer::StartConnServer(
  const char *listen_addr,
  unsigned short listen_port,
  const char *kvdb_path) {
  if (tcp_server_) {
    LOG(ERROR) << "Can't reinit the connect server";
    return false;
  }
  try {
    if(!kvdb_sqlite_->InitKvdb(kvdb_path)) {
      return false;
    }
    LOG(INFO) << "Start create local tcp server "
              << listen_addr << ":" << listen_port;
    tcp_server_ = vzconn::VzServer::CreateIpcServer(
                    *io_service_, listen_addr, listen_port);
    if (!tcp_server_) {
      LOG(ERROR) << "Failure to create tcp server";
    }
    tcp_server_->SignalNewConnect.connect(
      boost::bind(&KvdbServer::OnNewTcpConnect, shared_from_this(), _1, _2));
    tcp_server_->SignalVzServerError.connect(
      boost::bind(&KvdbServer::OnTcpServerError, shared_from_this(), _1, _2));
    // setting the transaction async timer
#ifdef WIN32
    timer_.reset(new boost::asio::deadline_timer(*io_service_));
#else
    timer_.reset(new boost::asio::steady_timer(*io_service_));
#endif
    WattingNextTransactionAsyncEvent();
    return tcp_server_->Start();
    LOG(INFO) << "Create Local tcp server succeed";
  } catch (std::exception &e) {
    LOG(ERROR) << e.what();
    return false;
  }
  return false;
}

void KvdbServer::StopConnServer() {
  LOG(INFO) << "Stop conn server";
  tcp_server_->Stop();
}

void KvdbServer::WattingNextTransactionAsyncEvent() {
#ifdef WIN32
  timer_->expires_from_now(boost::posix_time::seconds(ASYNC_TIMEOUT_TIMES));
#else
  timer_->expires_from_now(boost::chrono::seconds(ASYNC_TIMEOUT_TIMES));
#endif
  timer_->async_wait(
    boost::bind(&KvdbServer::OnHandleTimeout, shared_from_this(),
                boost::asio::placeholders::error));
}

void KvdbServer::OnHandleTimeout(const boost::system::error_code& err) {
  kvdb_sqlite_->ForceTransaction();
  WattingNextTransactionAsyncEvent();
}

void KvdbServer::OnNewTcpConnect(vzconn::VzServer::Ptr tcp_server,
                                 vzconn::VzConnect::Ptr tcp_connect) {
  LOG(INFO) << "New Connect " << tcp_connect->ip_addr();
  try {
    KvdbSession::Ptr kvdb_session(new KvdbSession(tcp_connect, kvdb_sqlite_));
    kvdb_session->SignalSessionError.connect(
      boost::bind(&KvdbServer::OnSessionError, shared_from_this(), _1, _2));
    kvdb_sessions_.push_back(kvdb_session);
    kvdb_session->StartSession();
  } catch (std::exception &e) {
    LOG(ERROR) << e.what();
  }
}

void KvdbServer::OnTcpServerError(vzconn::VzServer::Ptr tcp_server,
                                  const boost::system::error_code& err) {
  kvdb_sessions_.clear();
  // No need to stop tcp server
}

void KvdbServer::OnSessionError(KvdbSession::Ptr kvdb_session,
                                const boost::system::error_code& err) {
  kvdb_sqlite_->ForceTransaction();
  for(std::size_t i = 0; i < kvdb_sessions_.size(); i++) {
    if(kvdb_sessions_[i] == kvdb_session) {
      kvdb_sessions_.erase(kvdb_sessions_.begin() + i);
      return;
    }
  }
}

}